package com.lee.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.time.temporal.TemporalUnit;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * @Description TODO
 * @Author winston
 * @DateTime 2021/11/26
 */
public class MyConsumer {

	/**
	 * 主要
	 *
	 * @param args arg游戏
	 */
	public static void main(String[] args) {
		Properties init = init();
		KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(init);
		normalSubscribe(kafkaConsumer, List.of("testTopicNew2","testTopicNew1"));
		kafkaConsumer.close();
	}

	/**
	 * 初始化
	 *
	 * @return {@link Properties}
	 */
	public static Properties init() {
		Properties prop = new Properties();
		prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
		// 自动提交的延时
		prop.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
		prop.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
		// 反序列化
		prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
		prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

		prop.put(ConsumerConfig.GROUP_ID_CONFIG, "localhost_1");

		// 1. 新消费者组会从头开始消费
		// 2. 当前的offset位置的数据不存在了, 消费过的数据不存在了.
		// 3. 如何重新消费一个topic: 换组名+earliest
		prop.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");// earliest, latest
		return prop;
	}

	/**
	 * 正常的订阅
	 *
	 * @param consumer 消费者
	 * @param topics   主题
	 */
	public static void normalSubscribe(KafkaConsumer<String, String> consumer, List<String> topics) {
		consumer.subscribe(topics);
		while (true) {
			ConsumerRecords<String, String> poll = consumer.poll(Duration.of(1, ChronoUnit.SECONDS));
			poll.iterator().forEachRemaining(i -> {
				System.out.println(i.key()+"---"+i.value());
			});
//			consumer.commitSync();
			consumer.commitAsync();
		}
	}


}
